package com.song.hbaseChap_5_01;

import java.io.IOException;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;

public class ExampleMapper  extends TableMapper<Writable, Writable>{
	
	private Text k = new Text();
	private Text v = new Text();
	
	public static final String FIELD_COMMON_SEPARATOR="\u0001";
	
	@Override
	protected void map(
			ImmutableBytesWritable row,
			Result colume,
			Mapper<ImmutableBytesWritable, Result, Writable, Writable>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		
		String rowkey = new String(row.get());
		String value = null;
		byte[] columnFamily = null;
		byte[] columnQualifier = null;
		
		long ts = 0L;
		try {
			for(KeyValue kv :colume.list())
			{
				value = Bytes.toStringBinary(kv.getValue());
				columnFamily = kv.getFamily();
				columnQualifier = kv.getQualifier();
				ts = kv.getTimestamp();
				
				if("value1".equals(value));
				{
					k.set(rowkey);
					
					v.set(Bytes.toString(columnFamily) + FIELD_COMMON_SEPARATOR
							+Bytes.toString(columnQualifier)
							+FIELD_COMMON_SEPARATOR+value
							+FIELD_COMMON_SEPARATOR+ts);
					context.write(k, v);
					break;
				}
				
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
	}

	
	
	
	
}
